package com.shujia.util

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

object FlinkTableEnv {
  def getTableEnv(interval: Long): TableEnvironment = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
      * 创建flink sql环境
      *
      */
    val settings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .inStreamingMode() //流处理模式
      .build()

    val table: TableEnvironment = StreamTableEnvironment.create(env, settings)

    /**
      * 开启checkpoint
      *
      */
    FlinkCheckpoint.doCheckpoint(env, interval, "hdfs://master:9000/flink/car/checkpoint")


    //返回环境对象
    table
  }

}
